package com.dianmi.seckill.curator;

import com.dianmi.seckill.config.ZookeeperConfig;
import com.dianmi.seckill.entity.ProductStockFlag;
import com.dianmi.seckill.service.Callback;
import com.dianmi.seckill.service.StockCoordinate;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author xieqiaofu
 * @date 2021/4/2
 */
@Slf4j
@Component
public class CuratorCoordinate implements StockCoordinate {

    @Value("${server.port}")
    private Integer PORT = 8888;

    static String NET_ADDRESS = null;

    private static volatile CuratorFramework client = null;

    @Autowired
    private ZookeeperConfig zookeeperConfig;

    @PostConstruct
    public void init() {
        try {
            InetAddress address = InetAddress.getLocalHost();
            String hostAddress = address.getHostAddress();
            NET_ADDRESS = hostAddress + ":" + PORT;
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        if (client == null) {
            try {
                CuratorFramework client = CuratorFrameworkFactory.builder()
                        .connectString(zookeeperConfig.getConnectString())
                        .connectionTimeoutMs(zookeeperConfig.getConnectionTimeoutMs())
                        .sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMs())
                        .retryPolicy(new RetryNTimes(zookeeperConfig.getRetryTimes(), zookeeperConfig.getSleepBetweenRetries()))
                        .build();
                client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState connectionState) {
                        long sessionId = -1L;
                        try {
                            sessionId = client.getZookeeperClient().getZooKeeper().getSessionId();
                            log.info("zookeeper sessionId: " + sessionId);
                        } catch (Exception e) {
                            log.warn("Curator client state changed, but failed to get the related zk session instance.");
                        }

                        if (ConnectionState.CONNECTED == connectionState) {
                            log.info("Curator zookeeper client instance initiated successfully, session id is " + Long.toHexString(sessionId));
                        } else {
                            log.warn("Curator zookeeper state : {}", connectionState);
                        }
                    }
                });
                client.start();
                boolean connected = client.blockUntilConnected(10000, TimeUnit.MILLISECONDS);
                if (!connected) {
                    throw new IllegalStateException("zookeeper not connected");
                }
                CuratorCoordinate.client = client;
            } catch (Exception e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
        }
    }

    @Override
    public boolean createBaseNode() {
        if (client == null) {
            throw new IllegalStateException("zookeeper client is null");
        }
        try {
            Stat stat = client.checkExists().forPath(getBasePath());
            if (stat == null) {
                String baseNode = client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(getBasePath());
                log.info("baseNode: {}", baseNode);
                return true;
            }
        } catch (Exception e) {
            log.error("create base node failed", e);
        }
        return false;
    }

    @Override
    public boolean register(Long productId, Long ttl) {
        if (client == null) {
            throw new IllegalStateException("zookeeper client is null");
        }
        try {
            Stat stat = client.checkExists().forPath(getBasePath() + "/" + productId);
            if (stat == null) {
                String path = client.create()/*.withTtl(ttl)*/.creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(getBasePath() + "/" + productId, "1".getBytes());
                log.info("register product: {}" + path);
                return true;
            }
        } catch (Exception e) {
            log.error("register product error", e);
        }
        return false;
    }

    @Override
    public void listen(Callback callback) {
        if (client == null) {
            throw new IllegalStateException("zookeeper client is null");
        }
        try {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client, getBasePath(), true);
            pathChildrenCache.start();
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    if (PathChildrenCacheEvent.Type.CHILD_UPDATED == event.getType()) {
                        log.info("listen: {}, {}, {}", event.getType(), event.getData().getPath(), new String(event.getData().getData()));
                        if (null != callback) {
                            callback.call(event);
                        }
                    }
                }
            });
        } catch (Exception e) {
            log.error("listen error", e);
        }
    }

    @Override
    public void update(Long productId, boolean hasStock) {
        if (client == null) {
            throw new IllegalStateException("zookeeper client is null");
        }
        try {
            client.setData().forPath(getBasePath() + "/" + productId, hasStock ? "1".getBytes() : "2".getBytes());
        } catch (Exception e) {
            String errMsg = String.format("update product: {} data failed", productId);
            log.error(errMsg, e);
        }
    }

    @Override
    public List<ProductStockFlag> getProductStockFlag() {
        if (client == null) {
            throw new IllegalStateException("zookeeper client is null");
        }
        List<ProductStockFlag> list = null;
        try {
            List<String> paths = client.getChildren().forPath(getBasePath());
            if (!CollectionUtils.isEmpty(paths)) {
                list = new ArrayList<>();
                for (String path : paths) {
                    ProductStockFlag flag = new ProductStockFlag();
                    byte[] bytes = client.getData().forPath(getBasePath() + "/" + path);
                    String data = new String(bytes, StandardCharsets.UTF_8);
                    flag.setProductId(Long.valueOf(path));
                    flag.setData(data);
                    list.add(flag);
                }
            }
        } catch (Exception e) {
            log.error("init zk data failed", e);
        }
        return list;
    }

    private String getBasePath() {
        return "/seckill/stock";
    }
}
